package com.hhf.rrd.fraud_detection;

import com.hhf.rrd.fraud_detection.model.AccountDetail;
import com.hhf.rrd.fraud_detection.model.AlertInfo;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 实时诈骗检查任务
 *      开启socket端口：nc -l 9999
 *      查看端口占用: lsof -i :9999
 *
 * @author huanghaifeng15
 * @date 2022/2/11 16:57
 **/
public class FraudDetectionJob {
    public static void main(String[] args) throws Exception {
        // 获取运行上下文
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置 socket 数据源
        DataStreamSource<String> dataStreamSource = env.socketTextStream("localhost", 9999);

        // 对socket数据进行 transformation
        KeyedStream<AccountDetail, String> keyedStream = dataStreamSource.flatMap(new RichFlatMapFunction<String, AccountDetail>() {
            @Override
            public void flatMap(String value, Collector<AccountDetail> out) throws Exception {
                if (StringUtils.isNotBlank(value)) {
                    String[] wordArr = value.split(",");
                    int minArrLength = 2;
                    if (wordArr.length < minArrLength) {
                        return;
                    }
                    AccountDetail account = new AccountDetail();
                    account.setAccountId(wordArr[0]);
                    account.setAmount(Double.valueOf(wordArr[1]));
                    out.collect(account);
                }
            }
        }).map(new RichMapFunction<AccountDetail, AccountDetail>() {
            @Override
            public AccountDetail map(AccountDetail value) throws Exception {
                return value;
            }
        }).keyBy(new KeySelector<AccountDetail, String>() {
            @Override
            public String getKey(AccountDetail value) throws Exception {
                return value.getAccountId();
            }
        });
        keyedStream.print("fraud --> ");

        // 对欺诈数据进行清理
        SingleOutputStreamOperator<AlertInfo> processOperator = keyedStream.process(new FraudDetector());

        // sink alter
        processOperator.addSink(new AlertSink()).setParallelism(2);

        env.execute();
    }
}
